package control

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/antlinker/taskpool"

	"gopkg.in/mgo.v2"
)

type MgoTask struct {
	taskpool.BaseTask
	dbname   string
	collname string
	task     func(coll *mgo.Collection, opt ...interface{}) error
	opt      []interface{}
}

func (t *MgoTask) call(coll *mgo.Collection, opt ...interface{}) error {
	return t.task(coll, opt...)
}

//创建一个mongodb任务,指定集,合任务
func CreateTask(taskname, collname string, task func(coll *mgo.Collection, opt ...interface{}) error, opt ...interface{}) *MgoTask {
	return &MgoTask{
		collname: collname,
		task:     task,
		opt:      opt,
	}
}

//创建一个mongodb任务,指定数据库,集合,任务
func CreateDBTask(taskname, dbname, collname string, task func(coll *mgo.Collection, opt ...interface{}) error, opt ...interface{}) *MgoTask {

	return &MgoTask{
		dbname:   dbname,
		collname: collname,
		task:     task,
	}
}

const (
	taskName = "mongodb任务"
)

type myMgoOp struct {
	url               string
	session           *mgo.Session
	asyncTaskOperater taskpool.AsyncTaskOperater
	defaultDBName     string
}

var myMgo *myMgoOp

// GetMyMgo 获取数据库操作
func GetMyMgo() *myMgoOp {
	if myMgo == nil {
		panic("未初始化mongodb配置")
	}
	return myMgo
}

var lock = &sync.Mutex{}

// CreateMyMgo 创建mongodb数据库连接
func CreateMyMgo(mconfig *MongodbConfig) {
	lock.Lock()
	defer lock.Unlock()
	if myMgo == nil {
		myMgo = &myMgoOp{
			url:           mconfig.URL,
			defaultDBName: mconfig.DBName,
		}
		options := &taskpool.AsyncTaskOption{
			AsyncTaskMaxFaildedNum: mconfig.AsyncTaskMaxFaildedNum,
			AsyncMaxWaitTaskNum:    mconfig.AsyncMaxWaitTaskNum,
			MaxAsyncPoolNum:        mconfig.AsyncPoolNum,
			MinAsyncPoolNum:        1,
			AsyncPoolIdelTime:      time.Duration(mconfig.AsyncPoolIdelTime) * time.Second,
		}

		myMgo.asyncTaskOperater = taskpool.CreateAsyncTaskOperater("mongodb读写任务", myMgo, options)

		tmp, err := mgo.Dial(myMgo.url)
		if err != nil {
			panic(errors.New("创建mongodb连接失败:" + fmt.Sprintf("%v", err)))
		}
		myMgo.session = tmp
		// Optional. Switch the session to a monotonic behavior.
		myMgo.session.SetMode(mgo.Strong, true)

	}
}

func (m *myMgoOp) execTask(task *MgoTask) error {
	session, err := m.getSession()
	if err != nil {
		return err
	}
	defer session.Close()
	dbname := m.defaultDBName
	if task.dbname != "" {
		dbname = task.dbname
	}
	task.call(session.DB(dbname).C(task.collname), task.opt...)
	return nil
}

//同步执行
func (m *myMgoOp) ExecSync(collname string, task func(coll *mgo.Collection) error) error {
	session, err := m.getSession()
	if err != nil {
		Tlog.Debug("获取mongodb数据库会话失败:", err)
		return err
	}
	defer session.Close()
	return task(session.DB(m.defaultDBName).C(collname))
}

//异步执行
func (m *myMgoOp) ExecAsyncTask(task *MgoTask) {
	m.asyncTaskOperater.ExecAsyncTask(task)

}

//异步执行
func (m *myMgoOp) ExecDBAsync(db, collname string, task func(coll *mgo.Collection, opt ...interface{}) error, opt ...interface{}) {
	m.ExecAsyncTask(CreateDBTask(taskName, db, collname, task, opt...))
}

//异步执行
func (m *myMgoOp) ExecAsync(collname string, task func(coll *mgo.Collection, opt ...interface{}) error, opt ...interface{}) {
	m.ExecAsyncTask(CreateTask(taskName, collname, task, opt...))
}
func (m *myMgoOp) getSession() (*mgo.Session, error) {

	return m.session.Clone(), nil
}

func (m *myMgoOp) ExecTask(task taskpool.Task) error {
	mtask, ok := task.(*MgoTask)
	if ok {

		return m.execTask(mtask)
	}
	return nil
}
